package redis.gambing;

import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisLookupMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;

import static redis.gambing.GambleFinalString.GAMBLER_ID;
import static redis.gambing.GambleFinalString.HASH_GAMBLER_1_MONEY;

/**
 * 根据赌徒Id获取其当前赌资，然后发送到下游
 */
public class GamblerMoneyInfoLookupMapper implements RedisLookupMapper {
    private final RedisDataTypeDescription description;

    public GamblerMoneyInfoLookupMapper() {
        description = new RedisDataTypeDescription(RedisDataTypeDescription.RedisDataType.HASH, HASH_GAMBLER_1_MONEY);
    }

    @Override
    public List<Values> toTuple(ITuple iTuple, Object o) {
        String gamblerId = getKeyFromTuple(iTuple);
        List<Values> values = new ArrayList<>();
        values.add(new Values(gamblerId, Long.parseLong(String.valueOf(o))));
        return values;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        // 将赌徒id和该赌徒对应的赌资（HASH_GAMBLER_1_MONEY）发送出去
        outputFieldsDeclarer.declare(new Fields(GAMBLER_ID, HASH_GAMBLER_1_MONEY));
    }

    @Override
    public RedisDataTypeDescription getDataTypeDescription() {
        return description;
    }

    @Override
    public String getKeyFromTuple(ITuple iTuple) {
        return iTuple.getStringByField(GAMBLER_ID);
    }

    @Override
    public String getValueFromTuple(ITuple iTuple) {
        return null;
    }
}
